package com.lvmama.finance.demo.elasticsearch.sync;

import java.util.Date;
import java.util.List;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.stereotype.Component;

import com.lvmama.finance.demo.elasticsearch.domain.docs.ProductDoc;
import com.lvmama.finance.demo.elasticsearch.domain.jpa.ProductEntity;
import com.lvmama.finance.demo.elasticsearch.mapper.DozerMapper;
import com.lvmama.finance.demo.elasticsearch.repository.docs.ProductDocRepository;
import com.lvmama.finance.demo.elasticsearch.repository.jpa.ProductRepository;

@Component("productSyncModule")
public class ProductSyncModule implements SyncModule {

	@Autowired
	private ProductRepository productRepository;
	@Autowired
	private ProductDocRepository productDocRepository;
	@Autowired
	private DozerMapper mapper;
	
	private static final Integer BATCH_SIZE = 500;
	
	@Override
	public long doSync(Date lastSyncTime) {
		
		Page<ProductEntity> page = null;
		long total = 0;
		
		for (int pageNo=0 ;; pageNo++) {
			page = productRepository.findByModifiedTimeGreaterThanEqual(lastSyncTime, new PageRequest(pageNo, BATCH_SIZE));
			
			if (page.getContent().isEmpty()) 
				break;
			
			total += syncDataToEs(page.getContent());
			
			if (!page.hasNext()) {
				break;
			}
			
			try {
				Thread.sleep(1000);
			} catch (InterruptedException ignore) {
			}
			
		}
		return total;
	}
	
	private int syncDataToEs(List<ProductEntity> products) {
		List<ProductDoc> docs = mapper.mapToList(products, ProductDoc.class);
		productDocRepository.save(docs);
		return docs.size();
	}
}
